package com.zimmor.rocketmq.controller;

import com.zimmor.rocketmq.entity.RocketMessage;
import com.zimmor.rocketmq.selector.OrderSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @Author:zimmor
 * @Date: 2023/2/17 11:06
 * @Description
 * @version: 1.0.0
 */
@RestController
public class RocketMqController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送同步消息
     * @return
     */
    @GetMapping("/send")
    public String send() {
        for (int i = 0; i < 10; i++) {
            SendResult result = rocketMQTemplate.syncSend("base:tag", "BaseMessage--" + i);
            System.out.println("result = " + result);
        }
        return "OK" ;
    }

    /**
     * 发送异步消息
     * @return
     */
    @GetMapping("/send1")
    public String send1() {
        for (int i = 0; i < 10; i++) {
            rocketMQTemplate.asyncSend("base:tag", "BaseMessage--" + i, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("sendResult = " + sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    System.out.println("发送失败");
                    throwable.printStackTrace();
                }
            });
        }
        return "OK" ;
    }

    /**
     * 发送延时消息
     * @return
     */
    @GetMapping("/send2")
    public String send2() throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            final int index = i;
            rocketMQTemplate.asyncSend("base:tag", new Message<String>() {
                @Override
                public String getPayload() {
                    return "delayMessage" + index;
                }

                @Override
                public MessageHeaders getHeaders() {
                    Map<String, Object> map = new HashMap<>();
                    map.put("keys", "keys" + index);
                    MessageHeaders messageHeaders = new MessageHeaders(map);
                    return messageHeaders;
                }
            }, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                System.out.println(LocalDateTime.now() + "------------------------");
                    System.out.println("sendResult" + index + " = " + sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    System.out.println("发送失败");
                    throwable.printStackTrace();
                }
            }, 1500L, 3);
            TimeUnit.SECONDS.sleep(10);
        }
        return "OK" ;
    }

    /**
     * 批量消息
     * @return
     */
    @GetMapping("/send3")
    public String send3() {
        List<Message> list = new ArrayList<>();
//        Collection<com.zimmor.rocketmq.entity.Message> list = new ArrayList<>();
//        list.add(new Message() {
//            @Override
//            public Object getPayload() {
//                return "batchMessage01";
//            }
//
//            @Override
//            public MessageHeaders getHeaders() {
//                return null;
//            }
//        });
//        list.add(new Message() {
//            @Override
//            public Object getPayload() {
//                return "batchMessage02";
//            }
//
//            @Override
//            public MessageHeaders getHeaders() {
//                return null;
//            }
//        });
//        list.add(new Message() {
//            @Override
//            public Object getPayload() {
//                return "batchMessage03";
//            }
//
//            @Override
//            public MessageHeaders getHeaders() {
//                return null;
//            }
//        });
        list.add(new RocketMessage("batchMessage11"));
        list.add(new RocketMessage("batchMessage12"));
        list.add(new RocketMessage("batchMessage13"));
//        Collection<String> strings = Collections.synchronizedCollection(list);
        SendResult result = rocketMQTemplate.syncSend("base:tag", list);
        System.out.println("result = " + result);
        return "OK";
    }

    /**
     * 顺序消息
     * @return
     */
    @GetMapping("/send4")
    public String send4() {
        //设置队列选择器
        rocketMQTemplate.setMessageQueueSelector(new OrderSelector());
        for (int i = 0; i < 100; i++) {
            int orderId = i % 10;
            final int index = i;
//            RocketMessage rocketMessage = new RocketMessage("orderMessage" + i);
//            Map<String,Object> map = new HashMap<>();
//            map.put("keys", orderId + "");
//            MessageHeaders messageHeaders = new MessageHeaders(map);
//            rocketMessage.setMessageHeaders(messageHeaders);
            SendResult sendResult = rocketMQTemplate.syncSendOrderly("base:tag",  "orderMessage" + i, orderId + "");
            System.out.println("sendResult" + i + " = " + sendResult + "--" + i);
        }
        return "OK";
    }

    /**
     * 事务消息
     * @return
     */
    @GetMapping("/send5")
    public String send5() {
        TransactionSendResult transactionMessage = rocketMQTemplate.sendMessageInTransaction("base:tag", MessageBuilder.withPayload("TransactionMessage").build(), null);
        System.out.println("sendTesult" + transactionMessage);
        return "OK";
    }
}
